Add Bigtable CDC YAML SchemaTransform - #40261

This output schema looks complex, which will be noticed when using it with YAML.

In the regular BigtableReadSchemaTransformProvider, we added an option to use a flattened schema (one output per entry, instead of per mutation). Consider doing something similar here

Done

flatten: true breaks row-level atomicity and loses intra-MutateRow entry ordering:

In Bigtable, a MutateRow applies a list of mutations atomically to a single row, and per the MutateRowRequest contract, entries within a MutateRow are applied in order so later entries mask earlier ones (for example, deleteRow() / deleteFamily() followed by setCell() in an atomic row overwrite).

ChangeStreamMutation intentionally groups all entries of a row mutation together to preserve both row-level atomicity and entry ordering. Splitting mutation.getEntries() into independent PCollection<Row> elements breaks row-level atomicity and—because all entries in the same ChangeStreamMutation share the exact same key, commit_timestamp, and token with no sequence index—loses intra-mutation ordering across any downstream shuffle (so a DeleteFamily followed by a SetCell in the same MutateRow can be replayed in reverse order downstream).

How to fix (two options):

1. (Preferred) Remove flatten mode: Always output CHANGE_STREAM_MUTATION_SCHEMA (one Beam Row per ChangeStreamMutation with nested entries as ARRAY<ROW>), which preserves both row-level atomicity and entry ordering.
2. If flatten is needed for YAML convenience: Default flatten to false (opt-in rather than opt-out), and add an .addInt32Field("entry_index") (the 0-based index within mutation.getEntries()) to FLATTENED_CHANGE_STREAM_MUTATION_SCHEMA so downstream consumers can reconstruct the intra-mutation execution order.

Resource leak if setup or table cleanup fails:

1. table.create(), column_family.create(), update_table(), app_profile.create(), and row.commit() all run before the try ... finally: block (line 342). If any of those calls fail or time out after bigtable_instance.create() (line 285), finally: is never reached and the Bigtable instance is leaked.
2. Inside finally:, bigtable_instance.delete() is in the same try block after update_table() and table.delete(). If disabling the change stream or deleting the table raises an exception, bigtable_instance.delete() is skipped.
3. prefix='yaml_bt_cdc_it_' in the function signature is unused (instance_name = 'bt-cdc-tests' is hardcoded).

How to fix:
Move all post-instance-creation setup inside try:, and wrap bigtable_instance.delete() in its own try ... except block inside finally: so instance deletion is guaranteed even if table cleanup fails:

  table = bigtable_instance.table(table_id)
  table_name = f'projects/{project}/instances/{instance_id}/tables/{table_id}'
  try:
    table.create()
    # ... column_family, update_table, app_profile, row.commit(), yield ...
  finally:
    try:
      disable_request = bigtable_table_admin.UpdateTableRequest(
          table=table_pb.Table(name=table_name),
          update_mask=field_mask_pb2.FieldMask(paths=['change_stream_config']))
      bigtable_client.table_admin_client.update_table(
          request=disable_request).result(timeout=500)
      table.delete()
    except Exception:
      _LOGGER.warning('Failed to clean up Bigtable CDC table [%s]', table_id, exc_info=True)
    try:
      _LOGGER.info('Deleting Bigtable CDC instance [%s]', instance_id)
      bigtable_instance.delete()
    except Exception:
      _LOGGER.exception('Failed to delete Bigtable CDC instance [%s]', instance_id)

Do not expose withEndTime / end_at as a public user-facing option:

withEndTime is intentionally package-private and @VisibleForTesting ("Used only for integration tests. Unsafe to use in production."). Cloud Bigtable's low watermark is not guaranteed to be monotonic (it can go backwards), so terminating a change stream at endTime can lead to data loss—which is why withEndTime was purposefully disabled as a public option on BigtableIO.ReadChangeStream.

Making withEndTime public (which also triggers ErrorProne [VisibleForTesting] warnings when called from production code in another package) and exposing end_at / end_at_timestamp in the public ReadFromBigtableCDC YAML configuration exposes an unsafe API to users.

How to fix:
Revert BigtableIO.java and remove end_at_timestamp / end_at from BigtableChangeStreamReadConfiguration and standard_io.yaml.

Missing metadata table and ExistingPipelineOptions configuration:

BigtableChangeStreamReadConfiguration exposes change_stream_name and app_profile_id, but omits two sets of options needed in production Bigtable CDC pipelines:

1. existing_pipeline_options (withExistingPipelineOptions): BigtableIO.readChangeStream() defaults to ExistingPipelineOptions.FAIL_IF_EXISTS. If a user configures change_stream_name in YAML and restarts their pipeline, the job will fail on startup unless RESUME_OR_NEW (or SKIP_CLEANUP) can be configured.
2. Metadata table configuration (withMetadataTableProjectId, withMetadataTableInstanceId, withMetadataTableTableId, withMetadataTableAppProfileId, withCreateOrUpdateMetadataTable): Production deployments commonly store the metadata table in a separate Bigtable instance or use a transactional read-write app profile separate from a read-only source table/app profile.

How to fix:
Add optional configuration fields for existing_pipeline_options and metadata_table_* to BigtableChangeStreamReadConfiguration, wire them to BigtableIO.ReadChangeStream here, and expose their mappings in standard_io.yaml.

flatten: true breaks row-level atomicity and loses intra-MutateRow entry ordering:

In Bigtable, a MutateRow applies a list of mutations atomically to a single row, and per the MutateRowRequest contract, entries within a MutateRow are applied in order so later entries mask earlier ones (for example, deleteRow() / deleteFamily() followed by setCell() in an atomic row overwrite).

ChangeStreamMutation intentionally groups all entries of a row mutation together to preserve both row-level atomicity and entry ordering. Splitting mutation.getEntries() into independent PCollection<Row> elements breaks row-level atomicity and—because all entries in the same ChangeStreamMutation share the exact same key, commit_timestamp, and token with no sequence index—loses intra-mutation ordering across any downstream shuffle (so a DeleteFamily followed by a SetCell in the same MutateRow can be replayed in reverse order downstream).

How to fix (two options):

1. (Preferred) Remove flatten mode: Always output CHANGE_STREAM_MUTATION_SCHEMA (one Beam Row per ChangeStreamMutation with nested entries as ARRAY<ROW>), which preserves both row-level atomicity and entry ordering.
2. If flatten is needed for YAML convenience: Default flatten to false (opt-in rather than opt-out), and add an .addInt32Field("entry_index") (the 0-based index within mutation.getEntries()) to FLATTENED_CHANGE_STREAM_MUTATION_SCHEMA so downstream consumers can reconstruct the intra-mutation execution order.

Why bigtable_cdc.yaml times out here (and how to test this instead):

In BigtableIO.readChangeStream(), DetectNewPartitionsDoFn (DNP) and ReadChangeStreamPartitionDoFn (RCSP) are designed to run concurrently on a streaming runner: RCSP streams partitions and writes partition watermarks to the metadata table, while DNP tracks the global low watermark across all partitions in the metadata table and waits for all RCSPs to finish before terminating.

The Python YAML integration_tests.py harness runs external xlang transforms on Python's local batch FnApiRunner (fn_runner.py), which executes stages sequentially rather than concurrently—it buffers DNP's output in watermark_pending_inputs at MAX_TIMESTAMP and never starts RCSP until DNP finishes, creating a deadlock.

Since end_at shouldn't be exposed publicly on ReadFromBigtableCDC anyway (and FnApiRunner cannot run concurrent streaming SDF stages, which is also why ReadFromSpannerCDC and ReadFromPubSub don't run read pipelines in this harness), please remove bigtable_cdc.yaml / temp_bigtable_change_stream_table here and add a Java SchemaTransform integration test on a streaming runner instead (similar to SpannerChangeStreamsSchemaTransformIT / existing Bigtable CDC Java ITs).